微信开源mars源码分析5

2024-07-14 05:00| 来源: 网络整理| 查看: 265

最近回顾之前的文章，发现最后一篇有些着急了，很多地方没有叙述清楚。这里先做个衔接吧。我们还是以长连接为例，从longlink.cc看起。首先是那个线程函数__Run：/mars-master/mars/stn/src/longlink.cc

void LongLink::__Run() { ...... // 执行连接 SOCKET sock = __RunConnect(conn_profile); // 无效的socket，更新描述文件，记录失败的时间节点，返回 if (INVALID_SOCKET == sock) { conn_profile.disconn_time = ::gettickcount(); conn_profile.disconn_signal = ::getSignal(::getNetInfo() == kWifi); __UpdateProfile(conn_profile); return; } ...... // 执行读写 __RunReadWrite(sock, errtype, errcode, conn_profile); }

实际上核心的就2个，连接和读写，我们分别看下。/mars-master/mars/stn/src/longlink.cc

SOCKET LongLink::__RunConnect(ConnectProfile& _conn_profile) { std::vector ip_items; std::vector vecaddr; ...... // 赋值填充ip_items地址端口数组 netsource_.GetLongLinkItems(ip_items, dns_util_); ...... // 根据ip_items创建socket_address并加入vecaddr中 for (unsigned int i = 0; i < ip_items.size(); ++i) { vecaddr.push_back(socket_address(ip_items[i].str_ip.c_str(), ip_items[i].port).v4tov6_address(isnat64)); } ...... // 创建观察者和ComplexConnect连接核心，然后开始执行连接 LongLinkConnectObserver connect_observer(*this, ip_items); ComplexConnect com_connect(kLonglinkConnTimeout, kLonglinkConnInteral, kLonglinkConnInteral, kLonglinkConnMax); SOCKET sock = com_connect.ConnectImpatient(vecaddr, connectbreak_, &connect_observer); // 返回socket return sock; }

1.创建2个数组，地址端口item和socket_address；2.调用netsource_.GetLongLinkItems(ip_items, dns_util_);填充IPPortItem数组；3.根据填充好的前者数组生成socket_address填充后者数组;4.创建连接观察者;5.开始执行连接；首先看看netsource_.GetLongLinkItems是如何填充的：/mars-master/mars/stn/src/net_source.cc

bool NetSource::GetLongLinkItems(std::vector& _ipport_items, DnsUtil& _dns_util) { ScopedLock lock(sg_ip_mutex); if (__GetLonglinkDebugIPPort(_ipport_items)) { return true; } lock.unlock(); std::vector longlink_hosts = NetSource::GetLongLinkHosts(); if (longlink_hosts.empty()) { xerror2("longlink host empty."); return false; } __GetIPPortItems(_ipport_items, longlink_hosts, _dns_util, true); return !_ipport_items.empty(); }

可以看到debug的优先，这里增加了调试的ip。再往下就先不贴代码了，基本上就是之前通过SetLongLink设置进去的sg_longlink_hosts（长连接主机列表），再往上倒腾就是在MarsServiceNative.java的onCreate中通过描述文件profile设置进去的主机列表。也就是说之前早就设置好的主机列表已经存在了。下面我们仍然要进入到上一篇提到的ComplexConnect::ConnectImpatient这个核心函数中看看。/mars-master/mars/comm/socket/complexconnect.cc

SOCKET ComplexConnect::ConnectImpatient(const std::vector& _vecaddr, SocketSelectBreaker& _breaker, MComplexConnect* _observer) { ...... // 根据地址列表，生成ConnectCheckFSM连接列表 std::vector vecsocketfsm; for (unsigned int i = 0; i < _vecaddr.size(); ++i) { xinfo2(TSF"complex.conn %_", _vecaddr[i].url()); ConnectCheckFSM* ic = new ConnectCheckFSM(_vecaddr[i], timeout_, i, _observer); vecsocketfsm.push_back(ic); } // 下面就是对这个连接列表的各种操作了 do { ...... // 生成socketselect的封装对象，并执行PreSelect前期准备工作 SocketSelect sel(_breaker); sel.PreSelect(); ...... // 执行连接 for (unsigned int i = 0; i < index; ++i) { if (NULL == vecsocketfsm[i]) continue; xgroup2_define(group); vecsocketfsm[i]->PreSelect(sel, group); xgroup2_if(!group.Empty(), TSF"index:%_, @%_, ", i, this) Timeout()); } ...... for (unsigned int i = 0; i < index; ++i) { if (NULL == vecsocketfsm[i]) continue; xgroup2_define(group); vecsocketfsm[i]->AfterSelect(sel, group); xgroup2_if(!group.Empty(), TSF"index:%_, @%_, ", i, this) Status()) { if (_observer) _observer->OnFinished(i, socket_address(&vecsocketfsm[i]->Address()), vecsocketfsm[i]->Socket(), vecsocketfsm[i]->Error(), vecsocketfsm[i]->Rtt(), vecsocketfsm[i]->TotalRtt(), (int)(gettickcount() - starttime)); vecsocketfsm[i]->Close(); delete vecsocketfsm[i]; vecsocketfsm[i] = NULL; lasterror = -1; continue; } if (TcpClientFSM::EReadWrite == vecsocketfsm[i]->Status() && ConnectCheckFSM::ECheckFail == vecsocketfsm[i]->CheckStatus()) { if (_observer) _observer->OnFinished(i, socket_address(&vecsocketfsm[i]->Address()), vecsocketfsm[i]->Socket(), vecsocketfsm[i]->Error(), vecsocketfsm[i]->Rtt(), vecsocketfsm[i]->TotalRtt(), (int)(gettickcount() - starttime)); vecsocketfsm[i]->Close(); delete vecsocketfsm[i]; vecsocketfsm[i] = NULL; lasterror = -1; continue; } if (TcpClientFSM::EReadWrite == vecsocketfsm[i]->Status() && ConnectCheckFSM::ECheckOK == vecsocketfsm[i]->CheckStatus()) { if (_observer) _observer->OnFinished(i, socket_address(&vecsocketfsm[i]->Address()), vecsocketfsm[i]->Socket(), vecsocketfsm[i]->Error(), vecsocketfsm[i]->Rtt(), vecsocketfsm[i]->TotalRtt(), (int)(gettickcount() - starttime)); xinfo2(TSF"index:%_, sock:%_, suc ConnectImpatient:%_:%_, RTT:(%_, %_), @%_", i, vecsocketfsm[i]->Socket(), vecsocketfsm[i]->IP(), vecsocketfsm[i]->Port(), vecsocketfsm[i]->Rtt(), vecsocketfsm[i]->TotalRtt(), this); retsocket = vecsocketfsm[i]->Socket(); index_ = i; index_conn_rtt_ = vecsocketfsm[i]->Rtt(); index_conn_totalcost_ = vecsocketfsm[i]->TotalRtt(); vecsocketfsm[i]->Socket(INVALID_SOCKET); delete vecsocketfsm[i]; vecsocketfsm[i] = NULL; break; } } } while (true); }

1.数组中的每个长连接地址依次执行连接;2.数组中的每个连接分别做后续处理（一个for循环中的三段处理）;

我们首先看看vecsocketfsm[i]->PreSelect(sel, group);这句话,是由ConnectCheckFSM的父类TcpClientFSM实现的:/mars-master/mars/comm/socket/tcpclient_fsm.cc

void TcpClientFSM::PreSelect(SocketSelect& _sel, XLogger& _log) { switch(status_) { case EStart: { PreConnectSelect(_sel, _log); break; } case EConnecting: { _sel.Write_FD_SET(sock_); _sel.Exception_FD_SET(sock_); break; } case EReadWrite: { PreReadWriteSelect(_sel, _log); break; } default: xassert2(false, "preselect status error"); } }

这里是根据这个连接的当前状态决定前置操作的行为（开始连接、读写、连接中）。再往下看就是进行socket的connect。以PreConnectSelect为例，这里生产了socket，并执行了connect，最后将成功连接的socket执行_sel.Write_FD_SET(sock_);保存在了SocketSelect中。我们来看下代码：/mars-master/mars/comm/socket/tcpclient_fsm.cc

void TcpClientFSM::PreConnectSelect(SocketSelect& _sel, XLogger& _log) { xassert2(EStart == status_, "%d", status_); // 执行虚函数，由子类继承实现 _OnCreate(); xinfo2(TSF"addr:(%_:%_), ", addr_.ip(), addr_.port()) >> _log; // 生成socket sock_ = socket(addr_.address().sa_family, SOCK_STREAM, IPPROTO_TCP); if (sock_ == INVALID_SOCKET) { error_ = socket_errno; last_status_ = status_; status_ = EEnd; _OnClose(last_status_, error_, false); xerror2(TSF"close socket err:(%_, %_)", error_, socket_strerror(error_)) >> _log; return; } if (::getNetInfo() == kWifi && socket_fix_tcp_mss(sock_) < 0) { #ifdef ANDROID xinfo2(TSF"wifi set tcp mss error:%0", strerror(socket_errno)); #endif } if (0 != socket_ipv6only(sock_, 0)){ xwarn2(TSF"set ipv6only failed. error %_",strerror(socket_errno)); } if (0 != socket_set_nobio(sock_)) { error_ = socket_errno; xerror2(TSF"close socket_set_nobio:(%_, %_)", error_, socket_strerror(error_)) >> _log; } else { xinfo2(TSF"socket:%_, ", sock_) >> _log; } if (0 != error_) { last_status_ = status_; status_ = EEnd; return; } start_connecttime_ = gettickcount(); // 执行连接 int ret = connect(sock_, &(addr_.address()), addr_.address_length()); if (0 != ret && !IS_NOBLOCK_CONNECT_ERRNO(socket_errno)) { end_connecttime_ = ::gettickcount(); error_ = socket_errno; xwarn2(TSF"close connect err:(%_, %_), localip:%_", error_, socket_strerror(error_), socket_address::getsockname(sock_).ip()) >> _log; } else { xinfo2("connect") >> _log; // 记录socket到SocketSelect中 _sel.Write_FD_SET(sock_); _sel.Exception_FD_SET(sock_); } last_status_ = status_; if (0 != error_) status_ = EEnd; else status_ = EConnecting; if (0 == error_) _OnConnect(); }

需要注意的是_OnCreate的调用，实际上是子类实现的，这里也就是ConnectCheckFSM实现的：

virtual void _OnCreate() { if (m_observer) m_observer->OnCreated(m_index, addr_, sock_);}

这里将观察者与连接对象的生命周期绑在了一起，执行了观察者的OnCreated。那么观察者是谁呢？往上看，在LongLink::__RunConnect中生成的LongLinkConnectObserver。当然生命周期的回调并不止OnCreated一个。

回到__RunConnect中，看后续处理（for循环的三段处理）。执行AfterSelect并根据每个连接的状态决定后续处理，上篇已经讲过，不再累述。

那么何时终止这个do while循环呢？当for循环的三段处理完毕后，所有的连接过程都已经处理完毕了：

// end of loop bool all_invalid = true; for (unsigned int i = 0; i < vecsocketfsm.size(); ++i) { if (NULL != vecsocketfsm[i]) { all_invalid = false; break; } } if (all_invalid || INVALID_SOCKET != retsocket) break;

最后枚举一遍连接数组，每个元素检查是否非空，如果还有非空的，就将all_invalid置为false，那么会继续走一次do while。上面的三段处理完毕后，应该是数组中不再有连接才对，这里的保险处理是对数组再进行检查。至此跳出do while，算是整个连接过程完毕了。

可以看到，经过了三段处理后，连接数组中只会命中一个检测成功的连接，其他的失败和完成的都会置为null。这里从全局看就是一个地址池的淘汰筛选机制。在三段处理的for循环中清除不合格的连接，挑出第一个找到的合格的连接，然后跳出三段后，立刻检查整个数组是否已经就剩这一个可用了，如果不是继续执行do while，又会去执行数组中的每个item的连接过程，再回到三段处理。也就是说所有的数组中的item都会连接一次，然后根据返回的状态决定是否命中最终的一个socket。这是干嘛呢这么绕？我之前的理解恐怕还不透彻，现在感觉是在找一个稳定的可以读写状态的连接。第一次进入do while已经连接所有池中的item了，那么在经过了三段处理后淘汰掉不合适的和失败的，然后再进入do while再次执行vecsocketfsm[i]->PreSelect(sel, group);的时候，已经更新了状态并执行了不同的调用了，再经过三段处理在新的状态下再淘汰一批，最后经过整个运转，留下来的只能是最持久的（稳定的）唯一的一个连接，返回这个。不得不说，这里确实巧妙，如果我写并不会比这要好。

我们回来到longlink.cc的线程函数__Run中，当连接处理完毕后，下面继续执行的是__RunReadWrite。我们来看看：

void LongLink::__RunReadWrite(SOCKET _sock, ErrCmdType& _errtype, int& _errcode, ConnectProfile& _profile) { // Alarm消息触发处理绑定在__OnAlarm上 Alarm alarmnoopinterval(boost::bind(&LongLink::__OnAlarm, this), false); Alarm alarmnooptimeout(boost::bind(&LongLink::__OnAlarm, this), false); }

首先是2个Alarm，这里要理解就需要看看这个Alarm是个什么东西：/mars-master/mars/comm/alarm.h

template explicit Alarm(const T& _op, bool _inthread = true) : target_(detail::transform(_op)) , reg_async_(MessageQueue::InstallAsyncHandler(MessageQueue::GetDefMessageQueue())) , runthread_(boost::bind(&Alarm::__Run, this), "alarm") , inthread_(_inthread) , seq_(0), status_(kInit) , after_(0) , starttime_(0) , endtime_(0) , reg_(MessageQueue::InstallMessageHandler(boost::bind(&Alarm::OnAlarm, this, _1, _2), true)) #ifdef ANDROID , wakelock_(NULL) #endif {}

构造函数。这里需要逐句分析，首先是target_(detail::transform(_op))。简单看是个赋值语句，后面的参数需要看这个：/mars-master/mars/comm/thread/runnable.h

// base template for no argument functor template struct TransformImplement { static Runnable* transform(const T& t) { return new RunnableFunctor(t); } }; template inline Runnable* transform(const T& t) { return TransformImplement::transform(t); }

1.这里使用的是c++魔板，直接new了一个RunnableFunctor对象，这个对象是个runnable，其实就是将这个传递进来的参数t包装成了一个runnable，在适当的时候调用他的run方法的时候就会调用这个t了。那么带入到具体的内容中，这个t是_op，就是boost::bind(&LongLink::__OnAlarm, this)。这里又使用了c++的boost库，做了bind操作，绑定了参数this也就是LongLink与函数体LongLink::__OnAlarm。好了，现在target_是个包装好的runnable了，在适当的时候可以执行LongLink::__OnAlarm。

2.reg_async_(MessageQueue::InstallAsyncHandler(MessageQueue::GetDefMessageQueue()))。首先看MessageQueue::InstallAsyncHandler：/mars-master/mars/comm/messagequeue/message_queue.cc

MessageHandler_t InstallAsyncHandler(const MessageQueue_t& id) { ASSERT(0 != id); return InstallMessageHandler(__AsyncInvokeHandler, false, id); } MessageHandler_t InstallMessageHandler(const MessageHandler& _handler, bool _recvbroadcast, const MessageQueue_t& _messagequeueid) { ASSERT(bool(_handler)); ScopedLock lock(sg_messagequeue_map_mutex); const MessageQueue_t& id = _messagequeueid; if (sg_messagequeue_map.end() == sg_messagequeue_map.find(id)) { ASSERT2(false, "%" PRIu64, id); return KNullHandler; } HandlerWrapper* handler = new HandlerWrapper(_handler, _recvbroadcast, _messagequeueid, __MakeSeq()); sg_messagequeue_map[id].lst_handler.push_back(handler); return handler->reg; } struct HandlerWrapper { HandlerWrapper(const MessageHandler& _handler, bool _recvbroadcast, const MessageQueue_t& _messagequeueid, unsigned int _seq) : handler(_handler), recvbroadcast(_recvbroadcast) { reg.seq = _seq; reg.queue = _messagequeueid; } MessageHandler_t reg; MessageHandler handler; bool recvbroadcast; };

生成了一个HandlerWrapper，并将其保留在了一个map中，随后返回了MessageHandler_t，其中保存了_seq与_messagequeueid。这里我的感觉是这个handler就是个类似句柄的东西，保存MessageHandler的一个关联关系，即消息队列与seq码（这里是个自增的静态变量）。实际上调用者只要有这个MessageHandler_t就可以了。最后将这个MessageHandler_t赋值给了reg_async_。这里又有一个对象ScopeRegister是个MessageHandler_t的包装对象，里面统一封装了方法来操作MessageHandler_t。

3.runthread_(boost::bind(&Alarm::__Run, this), "alarm")。一个线程对象，线程函数是Alarm::__Run。没事什么好解释的。

4.inthread_(_inthread), seq_(0), status_(kInit), after_(0) , starttime_(0) , endtime_(0)。都是简单赋值，暂时不去管它。

5.reg_(MessageQueue::InstallMessageHandler(boost::bind(&Alarm::OnAlarm, this, _1, _2), true))。类似2。

好了，这个Alarm可以看做是个消息处理，在有消息触发的情况下会调用到具体的函数中，一般是__OnAlarm。

回到__RunReadWrite，往下看。首先是个while的死循环，我们单独摘录如下：

while (true) { ...... if (!alarmnoopinterval.IsWaiting()) { ...... if (__NoopReq(noop_xlog, alarmnooptimeout, has_late_toomuch)) { is_noop = true; __NotifySmartHeartbeatHeartReq(_profile, last_noop_interval, last_noop_actual_interval); } ...... } ...... // socket处理 SocketSelect sel(readwritebreak_, true); sel.PreSelect(); sel.Read_FD_SET(_sock); sel.Exception_FD_SET(_sock); ScopedLock lock(mutex_); if (!lstsenddata_.empty()) sel.Write_FD_SET(_sock); lock.unlock(); int retsel = sel.Select(10 * 60 * 1000); ...... // 处理发送(写入) if (sel.Write_FD_ISSET(_sock) && !lstsenddata_.empty()) { ...... ssize_t writelen = ::send(_sock, lstsenddata_.begin()->data.PosPtr(), lstsenddata_.begin()->data.PosLength(), 0); ...... while (it != lstsenddata_.end() && 0 < writelen) { if (0 == it->data.Pos()) OnSend(it->taskid); if ((size_t)writelen >= it->data.PosLength()) { xinfo2(TSF"sub send taskid:%_, cmdid:%_, %_, len(S:%_, %_/%_), ", it->taskid, it->cmdid, it->task_info, it->data.PosLength(), it->data.PosLength(), it->data.Length()) >> xlog_group; writelen -= it->data.PosLength(); if (!it->task_info.empty()) sent_taskids[it->taskid] = it->task_info; LongLinkNWriteData nwrite(it->taskid, it->data.PosLength(), it->cmdid, it->task_info); nsent_datas.push_back(nwrite); it = lstsenddata_.erase(it); } else { xinfo2(TSF"sub send taskid:%_, cmdid:%_, %_, len(S:%_, %_/%_), ", it->taskid, it->cmdid, it->task_info, writelen, it->data.PosLength(), it->data.Length()) >> xlog_group; it->data.Seek(writelen, AutoBuffer::ESeekCur); writelen = 0; } } } ...... // 处理接收(读取) if (sel.Read_FD_ISSET(_sock)) { bufrecv.AllocWrite(64 * 1024, false); ssize_t recvlen = recv(_sock, bufrecv.PosPtr(), 64 * 1024, 0); ...... while (0 < bufrecv.Length()) { uint32_t cmdid = 0; uint32_t taskid = Task::kInvalidTaskID; size_t packlen = 0; AutoBuffer body; int unpackret = longlink_unpack(bufrecv, cmdid, taskid, packlen, body); if (LONGLINK_UNPACK_FALSE == unpackret) { xerror2(TSF"task socket recv sock:%0, unpack error dump:%1", _sock, xdump(bufrecv.Ptr(), bufrecv.Length())); _errtype = kEctNetMsgXP; _errcode = kEctNetMsgXPHandleBufferErr; goto End; } xinfo2(TSF"task socket recv sock:%_, pack recv %_ taskid:%_, cmdid:%_, %_, packlen:(%_/%_)", _sock, LONGLINK_UNPACK_CONTINUE == unpackret ? "continue" : "finish", taskid, cmdid, sent_taskids[taskid], LONGLINK_UNPACK_CONTINUE == unpackret ? bufrecv.Length() : packlen, packlen); lastrecvtime_.gettickcount(); if (LONGLINK_UNPACK_CONTINUE == unpackret) { OnRecv(taskid, bufrecv.Length(), packlen); break; } else { sent_taskids.erase(taskid); bufrecv.Move(-(int)(packlen)); if (__NoopResp(cmdid, taskid, body, alarmnooptimeout, _profile)) { xdebug2(TSF"noopresp span:%0", alarmnooptimeout.ElapseTime()); is_noop = false; } else { OnResponse(kEctOK, 0, cmdid, taskid, body, _profile); } } } } } // 收尾，整个looper退出 End:

从while中的代码能够看出，基本上就是上面摘录的几块，如下所示：1.__NoopReq调用，无数据状态处理;2.socket的select处理;3.处理发送写入部分；4.处理接收读取部分；

这里需要逐个分析了：1.__NoopReq:先看代码，并不长：

bool LongLink::__NoopReq(XLogger& _log, Alarm& _alarm, bool need_active_timeout) { AutoBuffer buffer; uint32_t req_cmdid = 0; bool suc = false; if (identifychecker_.GetIdentifyBuffer(buffer, req_cmdid)) { suc = Send((const unsigned char*)buffer.Ptr(), (int)buffer.Length(), req_cmdid, Task::kLongLinkIdentifyCheckerTaskID); identifychecker_.SetSeq(Task::kLongLinkIdentifyCheckerTaskID); xinfo2(TSF"start noop synccheck taskid:%0, cmdid:%1, ", Task::kLongLinkIdentifyCheckerTaskID, req_cmdid) >> _log; } else { AutoBuffer body; longlink_noop_req_body(body); suc = SendWhenNoData((const unsigned char*) body.Ptr(), body.Length(), longlink_noop_cmdid(), Task::kNoopTaskID); xinfo2(TSF"start noop taskid:%0, cmdid:%1, ", Task::kNoopTaskID, longlink_noop_cmdid()) >> _log; } if (suc) { _alarm.Cancel(); _alarm.Start(need_active_timeout ? (2* 1000) : (10 * 1000)); } else { xerror2("send noop fail"); } return suc; }

说实话，这里看的不是很清晰，因为之前肯定有忽略的部分，我的猜测是在走了一个发送信令的校验后，根据返回的值的不同决定是执行send发送数据（使用校验填充好的buffer），还是走SendWhenNoData发送（自行填充请求体）没有数据的情况。暂时先往下看一步，看看Send：

bool LongLink::__Send(const unsigned char* _pbuf, size_t _len, uint32_t _cmdid, uint32_t _taskid, const std::string& _task_info) { lstsenddata_.push_back(LongLinkSendData()); lstsenddata_.back().cmdid = _cmdid; lstsenddata_.back().taskid = _taskid; longlink_pack(_cmdid, _taskid, _pbuf, _len, lstsenddata_.back().data); lstsenddata_.back().data.Seek(0, AutoBuffer::ESeekStart); lstsenddata_.back().task_info = _task_info; readwritebreak_.Break(); return true; }

这里能够清晰的看到，在使用lstsenddata_这个队列，来进行发送的请求，实际上就是向队列中增加一项。那么现在的问题就在于这个发送的数据时怎么来的了。这就需要我们弄懂LongLinkIdentifyChecker这个玩意儿。/mars-master/mars/stn/src/longlink_identify_checker.cc

bool LongLinkIdentifyChecker::GetIdentifyBuffer(AutoBuffer &_buffer, uint32_t &_cmdid) { if (has_checked_) return false; hash_code_buffer_.Reset(); _buffer.Reset(); IdentifyMode mode = (IdentifyMode)GetLonglinkIdentifyCheckBuffer(_buffer, hash_code_buffer_, (int&)_cmdid); switch (mode) { case kCheckNever: { has_checked_ = true; } break; case kCheckNext: { has_checked_ = false; } break; case kCheckNow: { cmd_id_ = _cmdid; return true; } break; default: xassert2(false); } return false; }

调用了GetLonglinkIdentifyCheckBuffer，我们追溯到stn_logic.cc中：

int GetLonglinkIdentifyCheckBuffer(AutoBuffer& identify_buffer, AutoBuffer& buffer_hash, int32_t& cmdid) { xassert2(sg_callback != NULL); return sg_callback->GetLonglinkIdentifyCheckBuffer(identify_buffer, buffer_hash, cmdid); }

实际上是对sg_callback这个回调的调用。最终我找到的线索是在MarsServiceNative.java上层的onCreate中设置了回调：

// set callback AppLogic.setCallBack(stub); StnLogic.setCallBack(stub); SdtLogic.setCallBack(stub);

再接着找到了MarsServiceStub.java中的getLongLinkIdentifyCheckBuffer：

@Override public int getLongLinkIdentifyCheckBuffer(ByteArrayOutputStream identifyReqBuf, ByteArrayOutputStream hashCodeBuffer, int[] reqRespCmdID) { // Send identify request buf to server // identifyReqBuf.write(); return ECHECK_NEVER; }

返回的是ECHECK_NEVER，没有填充buffer。也即是说has_checked_ = true，然后返回false。其实看到这一刻我是崩溃的，真心不知道是想干嘛。我们只能这么理解，只要进入__NoopReq其实都是在走SendWhenNoData发送无数据状态。好吧，我们重新回到__RunReadWrite中看一下。每次在while循环中一上来只要不是alarmnoopinterval正在等待的状态，那么就走一个发送无数据状态。看看发送无数据的代码：

bool LongLink::SendWhenNoData(const unsigned char* _pbuf, size_t _len, uint32_t _cmdid, uint32_t _taskid) { ScopedLock lock(mutex_); if (kConnected != connectstatus_) return false; if (!lstsenddata_.empty()) return false; return __Send(_pbuf, _len, _cmdid, _taskid, ""); }

其实是检查lstsenddata_是否有内容，如果没有才发送。那么好吧，这里整体理解就是每次whie循环开始都会检查如果发送队列中没有数据的时候，发送一个特定的无数据状态来确认连接。但是这里写的比较复杂，还需要回调回上层java的代码中，让其来控制状态，从而根据状态控制流程，只能说考虑的很周全，任何情况在任何节点都可以有处理。吐槽下如果我们自己写来规划这部分的时候大多数时候都是最对无数据检测放在下层，然后直接就发送了，不会让上层这里进行什么干涉吧。其实这里还有些点没有详细的分析很清楚，原谅文章有限，毕竟不能偏离主线太多。

2.socket的select操作。这里倒没什么可说的，前面的设置，为后面的sel.Select(10 60 1000)做准备，内部采用poll来运作。

3.发送过程。先是判断如果发送队列里面有内容，执行下面的::send(_sock, lstsenddata_.begin()->data.PosPtr(), lstsenddata_.begin()->data.PosLength(), 0)。这里注意，参数给定的是队列的第一个的data，也就是说这里是取出第一个执行发送。下面就是一个while循环，将发送队列过了一遍。如果刚才发送的数据大小与待发送的实际数据长度相等，那么认为是发送完了这一个，从队列中移除这一个，然后下一次while会自动取下一个了。如果没有；认为是没发完，位移数据，下次while仍然获取到这个item，但是数据位移已经变了，因此继续发送下面的数据。经过这个while之后，所有的发送队列中的数据都应当发送完毕了。

4.接收过程。前面没什么好说的，无非是开辟buffer空间，然后执行recv调用。之后进入一个while循环，条件是读取的buffer有数据。首先走一个解包调用，内部走的是__unpack_test，具体内容就不贴了，我简单看了下，基本上就是解开头部，头部的信息标识了本次传递的基本信息，包括了版本号等内容，一个结构体，还是比较标准的。这里是尝试解包，如果本次接收到的大小连头部都不够，那肯定返回错误，需要继续接收了。那么从这个能够看出，每次传递的数据都是带有一个头部的__STNetMsgXpHeader。这东西里面塞入的内容可以和客户端的版本，当前这个信令的id等关联起来。再下去看到的就是对解包返回值的判断了，如果一切顺利，就走到sent_taskids.erase(taskid);这里需要注意，这个sent_taskids是个发送的taskid的map，这里可以推测发送和接受其实是关联的，这里接收完毕移除这个保留项。然后走的__NoopResp这个调用。如果返回false表示不是空的信令返回，那么就走OnResponse。这个函数实际上是在LongLinkTaskManager中绑定了longlink_->OnResponse = boost::bind(&LongLinkTaskManager::__OnResponse, this, _1, _2, _3, _4, _5, _6);绑定在了LongLinkTaskManager::__OnResponse这里。

void LongLinkTaskManager::__OnResponse(ErrCmdType _error_type, int _error_code, uint32_t _cmdid, uint32_t _taskid, AutoBuffer& _body, const ConnectProfile& _connect_profile) { copy_wrapper body(_body); RETURN_LONKLINK_SYNC2ASYNC_FUNC(boost::bind(&LongLinkTaskManager::__OnResponse, this, _error_type, _error_code, _cmdid, _taskid, body, _connect_profile)); ...... int err_code = 0; int handle_type = Buf2Resp(it->task.taskid, it->task.user_context, body, err_code, Task::kChannelLong); switch(handle_type){ case kTaskFailHandleNoError: { dynamic_timeout_.CgiTaskStatistic(it->task.cgi, (unsigned int)it->transfer_profile.send_data_size + (unsigned int)body->Length(), ::gettickcount() - it->transfer_profile.start_send_time); __SingleRespHandle(it, kEctOK, err_code, handle_type, _connect_profile); xassert2(fun_notify_network_err_); fun_notify_network_err_(__LINE__, kEctOK, err_code, _connect_profile.ip, _connect_profile.port); } break; ...... } }

其实就2件事，通过Buf2Resp底层回包返回给上层解析。如果没有错误kTaskFailHandleNoError，会执行__SingleRespHandle：

bool LongLinkTaskManager::__SingleRespHandle(std::list::iterator _it, ErrCmdType _err_type, int _err_code, int _fail_handle, const ConnectProfile& _connect_profile) { ...... int cgi_retcode = fun_callback_(_err_type, _err_code, _fail_handle, _it->task, (unsigned int)(curtime - _it->start_task_time)); ...... }

这里的关键点就这一个，调用回调，回调的绑定在net_core.cc中的NetCore构造里，longlink_task_manager_->fun_callback_ = boost::bind(&NetCore::__CallBack, this, (int)kCallFromLong, _1, _2, _3, _4, _5);，最终执行的是NetCore::__CallBack:

int NetCore::__CallBack(int _from, ErrCmdType _err_type, int _err_code, int _fail_handle, const Task& _task, unsigned int _taskcosttime) { if (task_callback_hook_ && 0 == task_callback_hook_(_from, _err_type, _err_code, _fail_handle, _task)) { xwarn2(TSF"task_callback_hook let task return. taskid:%_, cgi%_.", _task.taskid, _task.cgi); return 0; } if (kEctOK == _err_type || kTaskFailHandleTaskEnd == _fail_handle) return OnTaskEnd(_task.taskid, _task.user_context, _err_type, _err_code); if (kCallFromZombie == _from) return OnTaskEnd(_task.taskid, _task.user_context, _err_type, _err_code); #ifdef USE_LONG_LINK if (!zombie_task_manager_->SaveTask(_task, _taskcosttime)) #endif return OnTaskEnd(_task.taskid, _task.user_context, _err_type, _err_code); return 0; }

看到了吧，走了OnTaskEnd，任务结束。

此文从中间部分开始粗糙了，前面铺垫的东西后面没有讲到，心不静的时候分析东西效果确实不大好。总而言之既然坚持写完了，这里还是留个记录吧，日后有机会的时候会回顾把这部分完善好。

【本文地址】

微信开源mars源码分析5

微信开源mars源码分析5

今日新闻

推荐新闻